#include <boost/asio.hpp>
#include <boost/asio/posix/stream_descriptor.hpp>
#include <iostream>
#include <thread>
#include <queue>
#include <mutex>
#include <sys/eventfd.h>
#include <unistd.h>
#include <atomic>

using namespace boost::asio;

/**
 * 高级eventfd示例：生产者-消费者模式
 * 
 * 这个例子展示了如何使用eventfd实现一个高效的生产者-消费者队列：
 * 1. 生产者线程向队列中添加数据，并通过eventfd通知消费者
 * 2. 消费者通过Asio异步等待eventfd事件，然后处理队列中的数据
 * 3. eventfd的值表示队列中待处理的数据数量
 */
template<typename T>
class EventfdQueue {
private:
    io_context& io_;
    posix::stream_descriptor eventfd_descriptor_;
    int eventfd_handle_;
    
    // 数据队列和同步原语
    std::queue<T> queue_;
    std::mutex queue_mutex_;
    std::atomic<bool> running_{true};
    
public:
    EventfdQueue(io_context& io) : io_(io), eventfd_descriptor_(io) {
        // 创建eventfd
        eventfd_handle_ = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if (eventfd_handle_ == -1) {
            throw std::runtime_error("Failed to create eventfd");
        }
        
        eventfd_descriptor_.assign(eventfd_handle_);
        std::cout << "EventFD队列创建成功" << std::endl;
    }
    
    ~EventfdQueue() {
        running_ = false;
        if (eventfd_handle_ != -1) {
            close(eventfd_handle_);
        }
    }
    
    // 生产者接口：添加数据到队列
    void produce(T item) {
        {
            std::lock_guard<std::mutex> lock(queue_mutex_);
            queue_.push(std::move(item));
        }
        
        // 通过eventfd通知有新数据
        uint64_t notification = 1;
        ssize_t result = write(eventfd_handle_, &notification, sizeof(notification));
        if (result != sizeof(notification)) {
            std::cerr << "EventFD通知失败" << std::endl;
        }
    }
    
    // 消费者接口：开始异步消费
    template<typename Handler>
    void start_consuming(Handler handler) {
        async_wait_and_consume(std::move(handler));
    }
    
    void stop() {
        running_ = false;
        // 发送停止信号
        produce(T{}); // 发送一个空项来唤醒消费者
    }
    
private:
    template<typename Handler>
    void async_wait_and_consume(Handler handler) {
        if (!running_) {
            return;
        }
        
        auto buffer = std::make_shared<uint64_t>();
        
        // 异步等待eventfd事件
        eventfd_descriptor_.async_read_some(
            boost::asio::buffer(buffer.get(), sizeof(*buffer)),
            [this, handler, buffer](const boost::system::error_code& ec, std::size_t bytes_read) {
                if (!running_) {
                    return;
                }
                
                if (!ec && bytes_read == sizeof(*buffer)) {
                    // 处理队列中的所有数据
                    process_queue_items(handler, *buffer);
                    
                    // 继续等待下一个事件
                    async_wait_and_consume(handler);
                } else if (ec != boost::asio::error::operation_aborted) {
                    std::cerr << "EventFD读取错误: " << ec.message() << std::endl;
                    
                    // 错误重试
                    std::this_thread::sleep_for(std::chrono::milliseconds(10));
                    async_wait_and_consume(handler);
                }
            }
        );
    }
    
    template<typename Handler>
    void process_queue_items(Handler& handler, uint64_t event_count) {
        std::lock_guard<std::mutex> lock(queue_mutex_);
        
        // 处理队列中的所有数据
        while (!queue_.empty()) {
            T item = std::move(queue_.front());
            queue_.pop();
            
            // 调用用户提供的处理函数
            handler(item);
        }
    }
};

// 任务数据结构
struct Task {
    int id;
    std::string description;
    
    Task() : id(0) {} // 默认构造函数用于停止信号
    Task(int i, std::string desc) : id(i), description(std::move(desc)) {}
};

void demonstrate_producer_consumer() {
    std::cout << "\n=== EventFD 生产者-消费者演示 ===" << std::endl;
    
    io_context io;
    EventfdQueue<Task> task_queue(io);
    
    // 消费者：处理任务的函数
    auto task_processor = [](const Task& task) {
        if (task.id == 0) {
            std::cout << "收到停止信号" << std::endl;
            return;
        }
        std::cout << "处理任务 [" << task.id << "]: " << task.description << std::endl;
        // 模拟任务处理时间
        std::this_thread::sleep_for(std::chrono::milliseconds(100));
    };
    
    // 启动消费者
    task_queue.start_consuming(task_processor);
    
    // 在单独线程中运行IO服务
    std::thread io_thread([&io]() {
        io.run();
    });
    
    // 生产者：生成任务
    std::thread producer([&task_queue]() {
        for (int i = 1; i <= 10; ++i) {
            Task task(i, "执行任务 #" + std::to_string(i));
            task_queue.produce(std::move(task));
            
            // 随机间隔生产任务
            std::this_thread::sleep_for(std::chrono::milliseconds(200 + (i % 3) * 100));
        }
        
        std::cout << "生产者完成所有任务生产" << std::endl;
    });
    
    // 等待生产者完成
    if (producer.joinable()) {
        producer.join();
    }
    
    // 等待一段时间确保所有任务都被处理
    std::this_thread::sleep_for(std::chrono::seconds(2));
    
    // 停止队列和IO服务
    task_queue.stop();
    io.stop();
    
    if (io_thread.joinable()) {
        io_thread.join();
    }
    
    std::cout << "生产者-消费者演示完成" << std::endl;
}

// 演示多生产者单消费者场景
void demonstrate_multi_producer() {
    std::cout << "\n=== 多生产者单消费者演示 ===" << std::endl;
    
    io_context io;
    EventfdQueue<std::string> message_queue(io);
    
    // 消费者
    auto message_processor = [](const std::string& message) {
        if (message.empty()) {
            return; // 停止信号
        }
        std::cout << "处理消息: " << message << std::endl;
    };
    
    message_queue.start_consuming(message_processor);
    
    // 启动IO线程
    std::thread io_thread([&io]() {
        io.run();
    });
    
    // 多个生产者线程
    std::vector<std::thread> producers;
    
    for (int producer_id = 1; producer_id <= 3; ++producer_id) {
        producers.emplace_back([&message_queue, producer_id]() {
            for (int i = 1; i <= 5; ++i) {
                std::string message = "生产者" + std::to_string(producer_id) + 
                                    " - 消息" + std::to_string(i);
                message_queue.produce(message);
                std::this_thread::sleep_for(std::chrono::milliseconds(150));
            }
        });
    }
    
    // 等待所有生产者完成
    for (auto& producer : producers) {
        if (producer.joinable()) {
            producer.join();
        }
    }
    
    // 清理
    std::this_thread::sleep_for(std::chrono::seconds(1));
    message_queue.stop();
    io.stop();
    
    if (io_thread.joinable()) {
        io_thread.join();
    }
    
    std::cout << "多生产者演示完成" << std::endl;
}

int main() {
    try {
        std::cout << "EventFD 高级使用示例" << std::endl;
        std::cout << "===================" << std::endl;
        
        demonstrate_producer_consumer();
        demonstrate_multi_producer();
        
    } catch (const std::exception& e) {
        std::cerr << "错误: " << e.what() << std::endl;
        return 1;
    }
    
    return 0;
}
